package link.luyu.plugin.brochain;

import io.reactivex.Flowable;
import io.reactivex.disposables.Disposable;
import java.math.BigInteger;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import link.luyu.plugin.brochain.types.BlockAndPage;
import link.luyu.plugin.brochain.web3.Web3;
import link.luyu.plugin.brochain.web3.methods.response.Block;
import link.luyu.plugin.brochain.web3.methods.response.Page;
import link.luyu.plugin.brochain.web3.methods.response.TransactionReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TransactionPolling {

    private static final Logger log = LoggerFactory.getLogger(TransactionPolling.class);

    private final String regionId;
    private final Set<String> consensusNodes;

    private final BrochainService service;
    private final Web3 web3;

    private final Cache cache;

    private final WeakHashMap<String, CompletableFuture<Optional<TransactionReceipt>>> futures;
    private final PriorityQueue<TransactionInfo> transactions;

    private Disposable polling;

    public TransactionPolling(
            String regionId, Set<String> consensusNodes, BrochainService service, Cache cache) {
        this.regionId = regionId;
        this.consensusNodes = consensusNodes;
        this.service = service;
        this.web3 = service.web3();
        this.cache = cache;
        this.futures = new WeakHashMap<>();
        this.transactions = new PriorityQueue<>(Comparator.comparing(a -> a.blockLimit));
    }

    private Flowable<BlockAndPage> blockPageFlowable() {
        return web3.blockNumberFlowable()
                .flatMap(
                        blockNumber ->
                                Flowable.fromFuture(service.getBlockAndPage(blockNumber, regionId)))
                .map(
                        blockAndPage -> {
                            if (Validator.validateBlockAndPage(blockAndPage, consensusNodes)) {
                                return blockAndPage;
                            } else {
                                throw new Exception("Invalid block");
                            }
                        });
    }

    private void startPolling() {
        if (polling == null || polling.isDisposed()) {
            polling =
                    blockPageFlowable()
                            .subscribe(
                                    this::processBlockAndPage,
                                    (exception) ->
                                            log.error(
                                                    "Error polling block: "
                                                            + exception.getMessage()));
        }
    }

    private void stopPolling() {
        polling.dispose();
    }

    public CompletableFuture<Optional<TransactionReceipt>> poll(
            String transactionHash, BigInteger blockLimit) {
        CompletableFuture<Optional<TransactionReceipt>> future;
        synchronized (futures) {
            future = futures.get(transactionHash);
            if (future == null) {
                future = new CompletableFuture<>();
                futures.put(transactionHash, future);
            }
        }
        synchronized (transactions) {
            transactions.add(new TransactionInfo(transactionHash, blockLimit));
        }
        startPolling();
        return future;
    }

    private void processBlockAndPage(BlockAndPage blockAndPage) {
        log.info(blockAndPage.toString());

        cache.updateLatestBlockNumber(blockAndPage.getBlock().getNumber());
        cache.addBlockAndPage(blockAndPage);

        Block block = blockAndPage.getBlock();
        Page page = blockAndPage.getPage();
        synchronized (futures) {
            if (page != null) {
                for (Page.PageTransaction transaction : page.getTransactions()) {
                    CompletableFuture<Optional<TransactionReceipt>> future =
                            futures.remove(transaction.getHash());
                    if (future != null) {
                        TransactionReceipt receipt =
                                new TransactionReceipt(
                                        transaction.getHash(),
                                        page.getHash(),
                                        page.getBlockNumberRaw(),
                                        page.getRegionId(),
                                        null, // TODO Can not get contract address
                                        transaction.getLogs());
                        future.complete(Optional.of(receipt));
                    }
                }
            }

            synchronized (transactions) {
                while (transactions.peek() != null
                        && transactions.peek().blockLimit.compareTo(block.getNumber()) <= 0) {
                    TransactionInfo transactionInfo = transactions.remove();
                    CompletableFuture<Optional<TransactionReceipt>> future =
                            futures.remove(transactionInfo.hash);
                    if (future != null) {
                        future.complete(Optional.empty());
                    }
                }
            }

            if (futures.isEmpty()) {
                stopPolling();
            }
        }
    }

    private static class TransactionInfo {

        private final String hash;
        private final BigInteger blockLimit;

        private TransactionInfo(String hash, BigInteger blockLimit) {
            this.hash = hash;
            this.blockLimit = blockLimit;
        }
    }
}
